package day03;


import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import untils.IntSource;

import java.util.ArrayList;
import java.util.Comparator;

public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .keyBy(r -> 1)
                .process(new KeyedProcessFunction<Integer, Integer, String>() {

                    private ListState<Integer> histryData;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        histryData = getRuntimeContext().getListState(
                                new ListStateDescriptor<Integer>("history", Types.INT));
                    }

                    @Override
                    public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
                        histryData.add(value);
                        ArrayList<Integer> integers = new ArrayList<>();
                        for (Integer i :histryData.get()) {
                            integers.add(i);
                        }

                        integers.sort(new Comparator<Integer>() {
                            @Override
                            public int compare(Integer o1, Integer o2) {
                                return o1 - o2;
                            }
                        });

                        StringBuilder result = new StringBuilder();
                        for (int i = 0; i < integers.size() -1 ; i++){
                            result.append(integers.get(i) + "->");
                        }

                        result.append(integers.get(integers.size() - 1));

                        out.collect(result.toString());
                    }
                })
                .print();

        env.execute();
    }
}
